package prototype.server;

import java.io.IOException;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.log4j.Level;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.beust.jcommander.JCommander;
import com.beust.jcommander.Parameter;
import com.beust.jcommander.ParameterException;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.ShutdownListener;
import com.rabbitmq.client.ShutdownSignalException;

/**
 * Base "Processor".<br>
 * Provides a {@link Connection} and a {@link Channel} and a single work queue
 * with a fetch size of one which must be ack-ed.
 */
public abstract class AmqpProcessor {
	private static final Logger _log = LoggerFactory.getLogger(AmqpProcessor.class);

	/*
	 * Command line arguments, annotation-based plumbing provided by
	 * com.beust.jcommander.JCommander
	 */
	@Parameter(names = "-help", description = "Display usage", help = true)
	private boolean _help = false;
	@Parameter(required = true, names = "-hostname", description = "AMQP Server Hostname")
	private String _hostname;
	@Parameter(required = false, names = "-port", description = "AMQP Server Port")
	private Integer _port = 5672;
	@Parameter(required = true, names = "-username", description = "Username to connect to AMQP")
	private String _username;
	@Parameter(required = true, names = "-password", description = "Password to connect to AMQP")
	private String _password;
	@Parameter(required = false, names = "-connectionTimeout", description = "Connection Timeout to AMQP Server in Seconds")
	private Integer _connectionTimeout = 5;
	@Parameter(required = false, names = "-heartbeatInterval", description = "Heartbeat Interval to AMQP Server in Seconds")
	private Integer _heartbeatInterval = 5;
	@Parameter(required = false, names = "-logLevel", description = "steve Log4j Log level")
	private String _logLevel = "INFO";
	// TODO: SSL with key/trust stores (JKS or PEM or P12)

	/**
	 * {@link ConnectionFactory} provides persistence for connection parameters
	 * and enables reconnects in case of abnormal {@link Channel} or
	 * {@link Connection} termination.
	 */
	private ConnectionFactory _factory = new ConnectionFactory();

	/**
	 * Each Processor has a single {@link Connection} and at least one
	 * {@link Channel}.
	 */
	protected Connection _connection;

	/**
	 * This {@link Channel} provides the main transport for operations. This
	 * class provides the constructs for the main work queue (acks + fetch of
	 * one).
	 */
	protected Channel _channel;

	/**
	 * Simple {@link ShutdownListener} that automatically tries to reconnect if
	 * a non-application-initiated shutdown was detected.
	 */
	protected ShutdownListener _shutdownListener = new ShutdownListener() {
		@Override
		public void shutdownCompleted(ShutdownSignalException signal) {
			if (!signal.isInitiatedByApplication()) {
				_log.error("Abnormal shutdown notification signal received: " + signal.getReason());
				startup();
			} else {
				_log.error("Normal shutdown notification signal received: " + signal.getReason());
			}
		}
	};

	/**
	 * App initiated shutdown sets this to true and prevents new connections.
	 */
	private AtomicBoolean isTerminated = new AtomicBoolean(false);

	/**
	 * Parse command line arguments and populate matching member variables.
	 * 
	 * @param args
	 *            {@link String} array of command line arguments.
	 */
	protected void parseCommandLineArguments(String[] args) {
		try {
			new JCommander(this, args);
		} catch (ParameterException e) {
			usage(e.getMessage());
		}
		if (_help)
			usage(null);
	}

	/**
	 * Print usage of command line arguments.
	 */
	protected void usage() {
		usage(null);
	}

	/**
	 * Print usage of command line arguments.
	 * 
	 * @param message
	 *            Optional {@link String} to be printed at end of usage
	 *            statement
	 */
	protected void usage(String message) {
		StringBuilder sb = new StringBuilder();
		new JCommander(this).usage(sb, "  ");
		if (message != null && message.trim().length() > 0)
			sb.append(message);
		System.err.println(sb.toString());
		System.exit(1);
	}

	/**
	 * Create a new AmqpProcessor using command line arguments.
	 * 
	 * @param args
	 *            {@link String} array of command line arguments.
	 */
	public AmqpProcessor(String[] args) {
		parseCommandLineArguments(args);

		org.apache.log4j.Logger.getLogger("prototype").setLevel(Level.toLevel(_logLevel));
		_factory.setHost(_hostname);
		_factory.setPort(_port);
		_factory.setUsername(_username);
		_factory.setPassword(_password);
		_factory.setConnectionTimeout(_connectionTimeout * 1000);
		_factory.setRequestedHeartbeat(_heartbeatInterval);
		startup();
	}

	/**
	 * Shutdown {@link Connection} and all of its {@link Channel}s.
	 * 
	 * @param wait
	 *            milliseconds to wait for outstanding operations to complete
	 *            before terminating.
	 */
	protected void shutdown(int wait) {
		_log.debug(String.format("AMQP Connection shutdown requested, waiting %s for outstanding requests to complete",
				(wait == 0 ? "forever" : wait + " milliseconds")));
		if (_connection != null) {
			_connection.removeShutdownListener(_shutdownListener);
			_connection.abort(wait);
		}
	}

	/**
	 * Shutdown {@link Connection} and all of its {@link Channel}s. Wait forever
	 * for outstanding operations to complete.
	 */
	protected void shutdown() {
		shutdown(0);
	}

	/**
	 * Call {@link AmqpProcessor#terminate(int)} and wait forever for
	 * outstanding operations to complete.
	 */
	public void terminate() {
		terminate(0);
	}

	/**
	 * Shutdown {@link Connection} and all of its {@link Channel}s and prevent
	 * any restarts. This is an application-initiated shutdown and essetially
	 * stops the processor.
	 * 
	 * @param wait
	 *            milliseconds to wait for outstanding operations to complete
	 *            before terminating.
	 */
	public void terminate(int wait) {
		isTerminated.set(true);
		shutdown(wait);
	}

	@Override
	protected void finalize() throws Throwable {
		shutdown(1000);
	}

	/**
	 * Create new {@link Connection}, connect and create single {@link Channel}.<br>
	 * Try forever until the {@link Connection} and {@link Channel} are
	 * initialized properly.<br>
	 * <br>
	 * Pause between reconnection attempts progressively back off:<br>
	 * <ul>
	 * <li>Attempts 1-10: 2 seconds between attempts
	 * <li>Attempts 11-20: 4 seconds between attempts
	 * <li>Attempts 21+: 10 seconds between attempts
	 * </ul>
	 * TODO: App initiated shutdown during startup should trump and terminate
	 * completely
	 */
	private void startup() {
		long attempts = 0;
		while (true && !isTerminated.get()) {
			try {
				_log.debug(String.format("Attempt %d,  connecting...", attempts++));
				shutdown(1);
				_connection = _factory.newConnection();
				_log.debug("Connection created : " + _connection.toString());
				_channel = _connection.createChannel();
				_log.debug("Channel created : " + _channel.toString());
				if (_connection.isOpen() && _channel.isOpen()) {
					_connection.addShutdownListener(_shutdownListener);
					_log.info("Connection successful: " + _connection.getServerProperties().toString());
					return;
				}
			} catch (IOException e) {
				_log.error("Failed to connect: " + e.getMessage());
			}

			try {
				int sleep = (attempts < 11 ? 2 : attempts < 21 ? 4 : 10);
				_log.debug(String.format("Waiting %d seconds before connecting again", sleep));
				Thread.sleep(sleep * 1000);
			} catch (InterruptedException e) {
			}
		}
	}

	protected abstract void configureWorkQueue(String exchange, String queue);

	public static void main(String[] args) throws Exception {
		String[] argv = { "-hostname", "jets", "-username", "guest", "-password", "guest", "-logLevel", "DEBUG" };
		AmqpProcessor p = new AmqpProcessor(argv) {
			@Override
			protected void configureWorkQueue(String exchange, String queue) {
			}
		};
		Thread.sleep(4000);
		p.terminate();
	}
}
